Using hasattr(data, 'expr') as a duck-typing check for PipelineVariable is fragile — any object with an expr attribute would match this condition. Since there's already a PipelineVariable isinstance check earlier in the function (lines 185-186), a more robust approach would be to repeat that isinstance check here, or better yet, import and check against the specific parameter types. This would also be more maintainable.

Alternatively, consider catching TypeError more broadly:

except TypeError:
    try:
        return str(data)
    except TypeError:
        # PipelineVariable.__str__ raises TypeError by design.
        # Return the object directly so pipeline serialization can handle it.
        return data

Returning data unconditionally in the inner except is safer than a duck-type check that could silently pass through unrelated objects. If the object truly can't be serialized or stringified, it's likely a PipelineVariable that should be passed through. If you want to be defensive, re-use the PipelineVariable isinstance check instead of hasattr.

Imports should be at the top of the file, not inside test functions. Move from sagemaker.train.utils import safe_serialize to the module-level imports. This applies to all three new test methods.

These tests in test_model_trainer_pipeline_variable.py are essentially duplicates of the tests already added in test_utils.py. The safe_serialize unit tests in test_utils.py are sufficient for testing the serialization behavior. These tests should instead verify the integration — that ModelTrainer correctly preserves PipelineVariable objects in its hyperparameters dict after construction, rather than just re-testing safe_serialize.

ParameterFloat and PipelineSession are imported but ParameterFloat is never used in the new tests, and PipelineSession is also unused. Remove unused imports.

The return type "str | PipelineVariable" is a forward reference string, but this module should use from __future__ import annotations at the top to enable PEP 604 union syntax properly. Also, this is a behavioral change to the function's contract — previously it always returned str, now it can return PipelineVariable. This could cause TypeError in downstream callers that expect a str (e.g., calling .encode(), concatenation, etc.). Have you audited all call sites of safe_serialize to ensure they handle a PipelineVariable return value correctly?

Suggestion: Add from __future__ import annotations at the module top, and change the annotation to:

def safe_serialize(data) -> str | PipelineVariable:

Also consider using Union[str, PipelineVariable] with an explicit import if from __future__ import annotations is not already used in this module.

The nested try/except is a reasonable defensive measure, but catching a bare TypeError from str(data) and silently returning the raw object is risky. This means any object whose __str__ raises TypeError (not just PipelineVariable) will be returned as-is, potentially causing unexpected behavior downstream. Consider being more explicit:

except TypeError:
    try:
        return str(data)
    except TypeError:
        if isinstance(data, PipelineVariable):
            return data
        raise

This way, only known PipelineVariable objects get the pass-through treatment, and truly broken objects still raise.

Missing type annotation for the data parameter. Per SDK conventions, all public functions must have type annotations for parameters and return types.

def safe_serialize(data: Any) -> str | PipelineVariable:

Per PEP 484 / SDK conventions, new modules should use from __future__ import annotations instead of from __future__ import absolute_import (which is a Python 2 artifact and unnecessary in Python 3).

Good test coverage overall! Consider adding a test for the specific edge case this PR is fixing — where isinstance check for PipelineVariable might fail (e.g., import path mismatch). You could mock isinstance or create a mock object whose __str__ raises TypeError to verify the nested except path:

def test_safe_serialize_with_object_whose_str_raises_typeerror():
    """Objects whose __str__ raises TypeError should be returned as-is."""
    class BadStr:
        def __str__(self):
            raise TypeError("cannot convert")
    obj = BadStr()
    result = safe_serialize(obj)
    assert result is obj

This directly tests the new fallback code path added in this PR.

These tests verify that ModelTrainer preserves PipelineVariable objects in the hyperparameters dict, but they don't test the actual serialization path (i.e., when safe_serialize is called during job creation). Consider adding a test that mocks the training job creation to verify that PipelineVariable values survive the full serialization pipeline, not just assignment.